import asyncio
import dataclasses
import json
import logging
import threading
import time
from enum import Enum
from logging import Logger
from typing import Optional, Dict, Set, List, Any, Callable

import websockets.exceptions
from websockets.legacy.client import connect
from websockets.legacy.protocol import WebSocketCommonProtocol

from .contants import OrderType, FutureType
from .entities import IndexPair, IndexPrice, Ticker, OrderBook, Instrument, SimpleInstrument, AuthInfo, CurrencySummary, \
    Position, Order, Trade, Future, Option
from ..entities import FutureFundRateDetail, DeliveryFundRateDetail, PerpetualFundRateDetail


class ExchangePublicServiceBase:
    platform = ''

    def __init__(self, env='test', logger: Logger = None, *args, **kwargs):
        self.env = env
        self._logger = logger or logging.root

    @property
    def name(self):
        return 'epubs'

    def init_quote_catcher(self, *args, **kwargs):
        raise NotImplementedError

    def index_price(self, pair: IndexPair) -> IndexPrice:
        raise NotImplementedError

    def mark_price(self, inst, **kwargs):
        raise NotImplementedError

    def ticker(self, inst_name: str, *args, **kwargs) -> Ticker:
        raise NotImplementedError

    def orderbook(self, inst_name: str, **kwargs) -> OrderBook:
        raise NotImplementedError

    def instruments_info(self, base_currency=None, margin_currency=None, kind=None, **kwargs) -> List[Instrument]:
        raise NotImplementedError

    def instrument_info(self, inst, **kwargs) -> Instrument:
        raise NotImplementedError

    @classmethod
    def parse_inst(cls, inst_name: str) -> Optional[SimpleInstrument]:
        raise NotImplementedError


class ExchangePriServiceBase:
    __instances = {}
    platform = ''

    def __init__(self, auth_info: AuthInfo, env='test', logger: Logger = None):
        self._auth_info = auth_info
        self.env = env
        self._logger = logger or logging.root

    @property
    def name(self):
        return 'epris'

    def check_auth(self, **kwargs):
        raise NotImplementedError

    def ensure_catcher_run(self):
        raise NotImplementedError

    def account_summaries(self, **kwargs) -> List[CurrencySummary]:
        """
        查询账户统计信息
        :return:
        """
        raise NotImplementedError

    def account_summary(self, base_currency=None, margin_currency=None) -> CurrencySummary:
        """
        查询账户对应币种统计信息
        :return:
        """
        raise NotImplementedError

    def positions(self, **kwargs) -> List[Position]:
        """
        查询持仓信息
        :param kwargs:
        :return:
        """
        raise NotImplementedError

    def open_orders(self, **kwargs, ) -> List[Order]:
        """
        查询挂单信息
        :param kwargs: 查询参数
        :return:
        """
        raise NotImplementedError

    def create_order(self, instrument, size, price, order_type=OrderType.Limit,
                     price_type=None, reduce_only=False, post_only=False, **kwargs):
        """
        创建订单
        :param instrument:
        :param size:
        :param price:
        :param order_type
        :param reduce_only
        :param price_type
        :param post_only
        :return:
        """
        raise NotImplementedError

    def reset_order(self, order_id, size=None, price=None, **kwargs):
        """
        创建订单
        :param order_id:
        :param size:
        :param price:
        :return:
        """
        raise NotImplementedError

    def cancel_order(self, order_id=None, label=None, **kwargs):
        """
        撤销订单
        :param order_id:
        :param label:
        :return:
        """
        raise NotImplementedError

    def order_trades(self, order_id, ) -> List[Trade]:
        """
        查询订单成交信息
        :param order_id:
        :return:
        """
        raise NotImplementedError

    def order_state(self, order_id, **kwargs):
        """
        根据订单ID查询订单信息
        :param order_id:
        :return:
        """
        raise NotImplementedError


class ExchangeError(Exception):
    def __init__(self, code: int, msg: str):
        self.code = code
        self.msg = msg

    def __str__(self):
        return f'code: {self.code}, msg: {self.msg}'


class ExchangeErrorCode(Enum):
    # 服务相关错误 (10000-10999)
    ServerUnavailable = 10001, 'Exchange server is unavailable'
    ServerRequestTimeout = 10002, 'Exchange request timeout'
    ServerRateLimit = 10003, 'Exceeded exchange request rate limit'
    ServerBusy = 10004, 'Exchange server is busy'
    ServerMaintenance = 10005, 'Exchange server is under maintenance'
    InvalidResponse = 10006, 'Invalid response from exchange'
    ServerDNSFailure = 10007, 'Exchange DNS resolution failed'
    ServerProtocolError = 10008, 'Protocol error in communication with exchange'
    ServerInternalError = 10009, 'Exchange server internal error'

    # 认证相关错误 (11000-11999)
    AuthFailed = 11001, 'Exchange authentication failed'
    IPWhiteListError = 11002, 'IP whitelist error'
    ApiKeyInvalid = 11003, 'API key is invalid'
    ApiKeyExpired = 11004, 'API key has expired'
    SignatureInvalid = 11005, 'Invalid signature'
    PermissionDenied = 11006, 'Permission denied for this API key'
    TokenExpired = 11007, 'Authentication token expired'
    TwoFactorAuthFailed = 11008, 'Two-factor authentication failed'
    InvalidCredentials = 11009, 'Invalid credentials provided'

    # 请求参数相关错误 (12000-12999)
    InvalidParameter = 12001, 'Invalid parameter'
    MissingParameter = 12002, 'Missing required parameter'
    InvalidSymbol = 12003, 'Invalid trading symbol'
    InvalidOrderType = 12004, 'Invalid order type'
    InvalidQuantity = 12005, 'Invalid order quantity'
    InvalidPrice = 12006, 'Invalid order price'
    InvalidTimestamp = 12007, 'Invalid or out-of-sync timestamp'
    ParameterOutOfRange = 12008, 'Parameter value out of range'
    InvalidFormat = 12009, 'Invalid parameter format'
    InvalidArguments = 12011, 'Invalid arguments provided'
    BadArgument = 12012, 'Bad argument passed'
    ExpiredSymbol = 12013, 'Expired or delisted trading symbol'
    UnsupportedSymbol = 12014, 'Symbol not supported in the current market'

    # 交易相关错误 (13000-13999)
    OrderNotExist = 13001, 'Order does not exist'
    MarginInsufficient = 13002, 'Insufficient margin'
    ExceedMaxPosition = 13003, 'Exceeded maximum position size'
    ExceedMaxOrderSize = 13004, 'Exceeded maximum order size'
    DuplicateOrderId = 13005, 'Duplicate client order ID'
    OrderRejected = 13006, 'Order rejected by exchange'
    OrderCanceled = 13007, 'Order has been canceled'
    OrderFilled = 13008, 'Order already filled'
    InvalidInstrument = 13009, 'Invalid instrument'
    PriceProtectionTriggered = 13010, 'Order rejected due to price protection mechanism'
    InvalidOrderStatus = 13011, 'Invalid order status for requested operation'
    MatchingFailed = 13012, 'Order matching failed'
    FrozenFundsInsufficient = 13013, 'Insufficient frozen funds for operation'
    SelfTradingRejected = 13014, 'Self-trading is not enabled'
    ActionOnClosedOrder = 13015, 'Action attempted on a closed order'
    CannotCancelLiquidationOrder = 13016, 'Cannot cancel a liquidation order'
    ComboLimitExceeded = 13017, 'Limit of active combo books reached'
    ShortOptionsLimitExceeded = 13018, 'Limit of total short options positions exceeded'
    PostOnlyOrderRejected = 13019, 'Post Only order rejected due to price modification not allowed'
    ResetOrderBySameParams = 13020, 'No need to modify the order'   # 使用相同参数修改订单

    # 账户相关错误 (14000-14999)
    AccountDisabled = 14001, 'Account is disabled'
    AccountInLiquidation = 14002, 'Account is in liquidation'
    AccountInRisk = 14003, 'Account risk level too high'
    WithdrawDisabled = 14004, 'Withdrawals are disabled'
    InsufficientBalance = 14005, 'Insufficient account balance'
    AccountFrozen = 14006, 'Account has been frozen'
    AssetNotSupported = 14007, 'Asset not supported for this account'
    InvalidWithdrawalAddress = 14008, 'Invalid withdrawal address'
    ContractTypeUnsupported = 14009, 'Unsupported contract type'
    NoDepositAddress = 14010, 'No deposit address specified'
    AccountLocked = 14011, 'Account is locked'

    # 网络与其他错误 (15000-15999)
    NetworkError = 15001, 'Network connection error'
    UnknownError = 15002, 'Unknown error occurred'
    OperationTimeout = 15003, 'Operation timed out'
    Retry = 15004, 'Request should be retried'

    def __call__(self, *args, **kwargs):
        return self.to_error(*args, **kwargs)

    @property
    def code(self):
        return self.value[0]

    def to_error(self, *args, **kwargs):
        code, msg = self.value
        return ExchangeError(code, msg)


@dataclasses.dataclass
class InstrumentQuote:
    inst_info: Instrument
    mark_price: Optional[float] = None
    index_price: Optional[float] = None
    interest_rate: Optional[float] = None

    def clear(self):
        self.mark_price = None
        self.index_price = None
        self.interest_rate = None


def generate_found_rate_by_quote(quote: InstrumentQuote, exchange) -> Optional[FutureFundRateDetail]:

    instrument = quote.inst_info.instrument_name
    currency = quote.inst_info.base_currency
    if not isinstance(quote.inst_info, Future):
        return
    is_reverse = quote.inst_info.reversed

    if quote.inst_info.future_type == FutureType.Perpetual:
        return PerpetualFundRateDetail(
            currency=currency, instrument=instrument, exchange=exchange, is_reverse=is_reverse,
            mark_price=quote.mark_price, index_price=quote.index_price,
            interest_rate=quote.interest_rate,
        )
    else:
        now = int(time.time())
        expire_ts = quote.inst_info.expire_timestamp // 1000
        duration = expire_ts - now
        found_rate = None
        if quote.mark_price and quote.index_price:
            found_rate = (quote.mark_price - quote.index_price) / quote.index_price / duration * 31536000

        return DeliveryFundRateDetail(
            currency=currency, instrument=instrument, exchange=exchange, is_reverse=is_reverse,
            mark_price=quote.mark_price, index_price=quote.index_price,
            found_rate=found_rate, expire_days=duration // 86400
        )


class AsyncioRunner:

    def __init__(self, *args, **kwargs):
        self._loop: Optional[asyncio.AbstractEventLoop] = None

    async def _run(self):
        raise NotImplementedError

    def run(self) -> threading.Thread:
        def _target():
            self._loop = asyncio.new_event_loop()
            asyncio.set_event_loop(self._loop)
            self._loop.run_until_complete(self._run())

        t = threading.Thread(target=_target, )
        t.setDaemon(True)
        t.start()
        return t

    async def run_as_coroutine(self):
        self._loop = asyncio.get_event_loop()
        return await self._run()


@dataclasses.dataclass
class AsyncRequest:
    id: Any
    payloads: Any
    callback: Optional[Callable]
    result: Any = None
    status: str = 'pending'

    async def set_result(self, result):
        if self.status != 'pending':
            raise ValueError('req is completed, can not repeat set result.')

        self.status = 'completed'
        self.result = result
        if self.callback and callable(self.callback):
            if asyncio.iscoroutinefunction(self.callback):
                await self.callback(result)
            else:
                self.callback(result)


class ExchangeWSRunner(AsyncioRunner):
    platform = ''

    def __init__(self, ws_url, connect_timeout=30, default_expire_ts=2000, logger=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        if not ws_url:
            logger.warning(f'init ws client for {self.name} without ws url, will ignore it.')

        self._logger: Logger = logger or logging.root
        self._ws_url = ws_url
        self._connect_timeout = connect_timeout
        self._client: Optional[WebSocketCommonProtocol] = None
        self._reconnect_times = 0
        self._subscribed_set: Set = set()
        self._last_actively_ts = 0
        self._base_request_id = 1
        self._request_lock = threading.Lock()
        self._requests = {}
        self._default_expire_ts = default_expire_ts
        self._is_run = False

    @property
    def name(self):
        return f'ws:{self.platform}'

    async def ws_url(self):
        return self._ws_url

    def is_run(self):
        return self._is_run

    def _should_be_run(self, connected=False):
        return True

    async def _run(self):
        if self._is_run:
            self._logger.info(f'{self.name} exchange ws runner is running, can not repeat')
            return

        self._is_run = True

        while self._should_be_run(connected=False):
            ws_url = None
            try:
                ws_url = await self.ws_url()
                if ws_url:
                    self._logger.info(f'{self.name} begin connect {self.platform} ws client with url {ws_url}')
                    self._client = await connect(
                        uri=ws_url, logger=self._logger, max_size=2 ** 20 * 4,
                        open_timeout=self._connect_timeout, ping_interval=10, ping_timeout=3
                    )
                    self._reconnect_times = 0
            except websockets.exceptions.WebSocketException as e:
                self._logger.error(f'{self.name} connect to {self._ws_url} failed: {e}')
            except Exception as e:
                self._logger.error(f'{self.name} connect to {self._ws_url} except: {e}')

            connected, message = False, None
            if self._client and not self._client.closed:
                self._logger.info(f'{self.name} websocket client connect to {ws_url} success')
                asyncio.ensure_future(self._on_connected())
                connected = True

            while self._client and not self._client.closed and self._should_be_run(connected=True):
                try:
                    message = await asyncio.wait_for(self._client.recv(), 2)
                    self._last_actively_ts = int(time.time() * 1000)
                    asyncio.ensure_future(self._on_message(json.loads(message)))
                except asyncio.TimeoutError:
                    pass
                except websockets.exceptions.WebSocketException as e:
                    self._logger.error(f'{self.name} ws connection got websocket exception {e}')
                except json.JSONDecodeError:
                    self._logger.warning(f'{self.name} got invalid message: {message}')
                    continue
                except Exception as e:
                    self._logger.exception(f'{self.name} handle ws message except: {e}')
            try:
                if connected:
                    await self._on_disconnect()
            except Exception as e:
                self._logger.exception(f'{self.name} handle disconnect event except: {e}')

            if self._should_be_run():
                delay_time = min(2 ** self._reconnect_times + 5, 300)
                self._logger.info(f'{self.name} will reconnect after {delay_time} '
                                  f'seconds, retry times is {self._reconnect_times}')
                await asyncio.sleep(delay_time)
                self._reconnect_times += 1
            else:
                break
        self._is_run = False

    @property
    def _request_id(self) -> int:
        with self._request_lock:
            request_id = self._base_request_id
            self._base_request_id += 1
        return request_id

    async def _on_connected(self):
        raise NotImplementedError

    async def _on_disconnect(self):
        self._logger.warning(f'{self.name} disconnect')
        if not self._client.closed:
            try:
                await self._client.close()
            except Exception as e:
                self._logger.warning(f'close client failed {e}')
        self._subscribed_set.clear()
        self._requests.clear()

    async def _on_message(self, message):
        raise NotImplementedError

    async def _request(self, *args, **kwargs) -> AsyncRequest:
        raise NotImplementedError

    def is_available(self, ):
        return self._loop and self._loop.is_running() and self._client and not self._client.closed


class ExchangeQuotesCatcher:
    platform = ''

    def __init__(self, *args, **kwargs):
        self._quotes: Dict[str, InstrumentQuote] = {}
        self._index_prices: Dict[str, IndexPrice] = {}

    def get_index_price(self, pair, ):
        raise NotImplementedError

    def get_instruments(self, ) -> List[Instrument]:
        if self._quotes:
            results = []
            now = int(time.time() * 1000)
            for val in self._quotes.values():
                inst_info = val.inst_info
                if isinstance(inst_info, (Future, Option)):
                    if inst_info.expire_timestamp and now > inst_info.expire_timestamp:
                        continue
                results.append(inst_info)
            return results

    def get_futures_info(self, currency=None) -> List[Future]:
        results = []
        for quote in self._quotes.values():
            if isinstance(quote.inst_info, Future):
                results.append(quote.inst_info)
        return results

    def get_future_found_rate_details(self, ) -> List[FutureFundRateDetail]:
        results = []
        for instrument, quote in self._quotes.items():
            if not quote.inst_info or not isinstance(quote.inst_info, Future):
                continue
            found_rate = generate_found_rate_by_quote(quote, self.platform)
            if found_rate:
                results.append(found_rate)
        return results


class ExchangeUserDataCatcher:
    def __init__(self, *args, **kwargs):
        """"""

    def run(self):
        raise NotImplementedError

    def account_summaries(self, *args, **kwargs):
        raise NotImplementedError

    def positions(self, *args, **kwargs):
        raise NotImplementedError

    def open_orders(self, *args, **kwargs):
        raise NotImplementedError
